package com.pk.flink.sink;


import com.pk.flink.bean.Access;
import com.pk.flink.functions.sink.AccessConsoleSinkFunction;
import com.pk.flink.functions.source.AccecssMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.time.Duration;

/**
 * 使用Flink 实现单词计数
 * @author pk
 */
public class FlinkRedisSinkApp {
    public static void main(String[] args) throws Exception {

        Configuration config = new Configuration();
        // 网络缓冲区占TaskManager内存的比例（默认0.1，调高至0.3）
        config.setFloat("taskmanager.memory.network.fraction", 0.3f);
        // 网络缓冲区最小内存（默认64MB，可设为128MB）
        config.setString("taskmanager.memory.network.min", "256mb");
        // 网络缓冲区最大内存（默认1GB，可设为256MB）
        config.setString("taskmanager.memory.network.max", "256mb");

        //获取上下文环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
        DataStreamSource<String> accessLogStreamSource = env.readTextFile("data/access.log");

        //Console Sink 打印到标准输出
        SingleOutputStreamOperator<Access> mapResults = accessLogStreamSource.map(new AccecssMapFunction());
        SingleOutputStreamOperator<Tuple2<String, Double>> map = mapResults
                .map(access -> new Tuple2<>(access.getDomain(), access.getTraffic()))
                .returns(Types.TUPLE(Types.STRING, Types.DOUBLE))
                .keyBy(t->t.f0)
                .sum(1);

        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").setDatabase(0).setPort(6379).build();

        map.addSink(new RedisSink<Tuple2<String, Double>>(conf, new RedisExampleMapper()));

        env.execute();
    }

    public static class RedisExampleMapper implements RedisMapper<Tuple2<String, Double>> {

        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET, "access_redis");
        }

        @Override
        public String getKeyFromData(Tuple2<String, Double> data) {
            return data.f0;
        }

        @Override
        public String getValueFromData(Tuple2<String, Double> data) {
            return data.f1 + "";
        }
    }

}
